package cn.lgwen.candy.runtime.core;

import cn.lgwen.candy.runtime.core.Graph.ProcessorDriversChain;
import cn.lgwen.candy.runtime.core.Graph.sourceDriverChain;
import cn.lgwen.candy.runtime.driver.Driver;
import cn.lgwen.candy.runtime.driver.ProcessorDriver;
import cn.lgwen.candy.runtime.driver.SinkDriver;
import cn.lgwen.candy.runtime.driver.SourceDriver;
import cn.lgwen.candy.runtime.operator.SourceOperator;
import cn.lgwen.candy.runtime.queue.MemoryBlockQueue;
import cn.lgwen.candy.runtime.queue.Queue;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 */
public class EnvironmentImpl implements Environment {

    private List<SourceStream<?>> sourceStreams;

    private boolean checkpoint;

    private long checkpointInterval;

    public void checkpoint(long millisecond) {
        if (millisecond <= 0) {
            return;
        }
        checkpoint = true;
        checkpointInterval = millisecond;
    }

    Environment create() {
        EnvironmentImpl environment =  new EnvironmentImpl();
        environment.sourceStreams = new LinkedList<>();
        return environment;
    }

    @Override
    public <O> Stream<O> addSource(SourceOperator<O> sourceOperator) {
        SourceStream <O> sourceStream = new SourceStream<>(this);
        sourceStream.setSourceOperator(sourceOperator);
        sourceStreams.add(sourceStream);
        return sourceStream;
    }

    @Override
    public void executor() throws Exception {
        Graph graph = createGraph();
        List<sourceDriverChain<?>> sourceChains = graph.getSourceChains();
        List<Runnable> tasks = new LinkedList<>();
        //每个source 表示一个流
        for (sourceDriverChain sourceChain : sourceChains) {
            SourceDriver sourceDriver = sourceChain.getSourceDriver();
            addTasks(tasks, sourceDriver);
            List<ProcessorDriversChain> processorDriversChains = sourceChain.getProcessorDriversChains();
            if (!processorDriversChains.isEmpty()) {
                for (ProcessorDriversChain processorDriversChain : processorDriversChains) {
                    Queue queue = new MemoryBlockQueue<>();
                    sourceDriver.registerForward(queue);
                    ProcessorDriver processorDriver = processorDriversChain.getProcessorDriver();
                    processorDriver.setUpstreamQueue(queue);
                    addTasks(tasks, processorDriver);
                    if (processorDriversChain.getSinkDriver() != null) {
                        Queue<?> memoryBlockQueue = new MemoryBlockQueue<>();
                        processorDriver.registerForward(memoryBlockQueue);
                        processorDriversChain.getSinkDriver().setUpstreamQueue(memoryBlockQueue);
                        addTasks(tasks, processorDriversChain.getSinkDriver());
                    }
                    if (!processorDriversChain.getProcessorDriversChains().isEmpty()) {
                        for (Object driversChain : processorDriversChain.getProcessorDriversChains()) {
                            ProcessorDriversChain processorDriversChain1 = (ProcessorDriversChain) driversChain;
                            Queue<?> memoryBlockQueue = new MemoryBlockQueue<>();
                            processorDriver.registerForward(memoryBlockQueue);
                            registerProcessorDriver(tasks, memoryBlockQueue, processorDriversChain1);
                        }
                    }
                }
            }
        }

        ExecutorService executors = Executors.newFixedThreadPool(tasks.size());
        for (Runnable runnable : tasks) {
            executors.execute(runnable);
        }
        while (!executors.isTerminated()) {
            TimeUnit.MILLISECONDS.sleep(10);
        }
        executors.shutdown();
    }

    public void registerProcessorDriver(List<Runnable> tasks, Queue upstreamQueue, ProcessorDriversChain<?> chain) {
        ProcessorDriver driver = chain.getProcessorDriver();
        driver.setUpstreamQueue(upstreamQueue);
        addTasks(tasks, driver);
        if (chain.getSinkDriver() != null) {
            chain.getSinkDriver().setUpstreamQueue(upstreamQueue);
            addTasks(tasks, chain.getSinkDriver());
        }
        if (!chain.getProcessorDriversChains().isEmpty()) {
            for (ProcessorDriversChain<?> driversChain : chain.getProcessorDriversChains()) {
                Queue<?> queue = new MemoryBlockQueue<>();
                driver.registerForward(queue);
                registerProcessorDriver(tasks, queue, driversChain);

            }
        }
    }

    void addTasks(List<Runnable> tasks, Driver driver) {
        driver.checkpoint(checkpoint);
        for (int i = 0; i < driver.parallel(); i++) {
            tasks.add(driver::run);
        }
    }


    public Graph createGraph() {
       // 从source 开始构建
        if (sourceStreams.isEmpty()) {
            throw new RuntimeException("no source operator defined");
        }
        Graph graph = new Graph();
        for (SourceStream<?> sourceStream : sourceStreams) {
            SourceOperator<?> sourceOperator = sourceStream.getSourceOperator();
            SourceDriver<?> sourceDriver = new SourceDriver<>(sourceOperator);
            sourceDriverChain<?> sourceDriverChain = new sourceDriverChain<>(sourceDriver);
            graph.addSourceChains(sourceDriverChain);
            if (!sourceStream.getNextStreams().isEmpty()) {
                for (Stream<?> nextStream: sourceStream.getNextStreams()) {
                    ProcessorDriversChain nextChain = new ProcessorDriversChain();
                    nextChain.setProcessorDriver(new ProcessorDriver(nextStream.processor));
                    sourceDriverChain.addProcessorDriver(nextChain);
                    processorDriversChain(nextChain, nextStream);
                }
            }
        }
        return graph;
    }
    /*
     * T 表示Stream 的输入类型，同sinkDriver 的输入类型
     */
    private ProcessorDriversChain processorDriversChain(ProcessorDriversChain<?> chain, Stream<?> stream) {
        List<Stream<?>> nextOperators = stream.nextStreams;
        if (stream.sinkOperator != null) {
            chain.setSinkDriver(new SinkDriver(stream.sinkOperator));
        }
        if (nextOperators != null && !nextOperators.isEmpty()) {
            for (Stream<?> nextStream: nextOperators) {
                ProcessorDriversChain nextChain = new ProcessorDriversChain();
                nextChain.setProcessorDriver(new ProcessorDriver(nextStream.processor));
                chain.addProcessorDriverChain(processorDriversChain(nextChain, nextStream));
            }
        }
        return chain;
    }

}
